package org.example.interceptor;


import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;

/**
 * 使用flume 判断 卡法卡 走那个主题 自定义拦截器
 * flum 自定义拦截器
 * 定义类型拦截器
 */
public class TypeInterceptor implements Interceptor {

    //声明一个集合、用于存放拦截器处理后的事件
    private List<Event> addHeaderEvents;

    @Override
    public void initialize() {
        //初始化集合用于存放拦截器处理后的事件
        addHeaderEvents = new ArrayList<>();
    }


    /**
     * 单个事件处理方法
     * event 包含body 和header
     * https://flume.apache.org/releases/content/1.9.0/FlumeUserGuide.html
     * @param event
     * @return
     */
    @Override
    public Event intercept(Event event) {
        //1、获取事件中的头部信息 header & body
        Map<String, String> headers = event.getHeaders();

        //获取事件中的body信息
        String body = new String(event.getBody());

        //根据body中是否有hello 来决定添加怎样的头部信息
        if (body.contains("hello")) {
            headers.put("topic", "first");
        } else {
            headers.put("topic", "second");
        }
        //返回数据
        return event;
    }


    /**
     * 批量事件处理方法
     *
     * @param list
     * @return
     */
    @Override
    public List<Event> intercept(List<Event> list) {
        //清空集合
        addHeaderEvents.clear();
        for (Event event : list) {
            //交给单个Event 处理
            addHeaderEvents.add(intercept(event));
        }

        //返回数据
        return addHeaderEvents;
    }

    @Override
    public void close() {


    }


    /**
     * a1.sources = r1
     * a1.sinks = k1
     * a1.channels = c1
     * a1.sources.r1.interceptors = i1 i2
     * a1.sources.r1.interceptors.i1.type = org.apache.flume.interceptor.HostInterceptor$Builder
     * a1.sources.r1.interceptors.i1.preserveExisting = false
     * a1.sources.r1.interceptors.i1.hostHeader = hostname
     * a1.sources.r1.interceptors.i2.type = org.apache.flume.interceptor.TimestampInterceptor$Builder
     * a1.sinks.k1.filePrefix = FlumeData.%{CollectorHost}.%Y-%m-%d
     * a1.sinks.k1.channel = c1
     * 帮助构建拦截器对象  $Builder
     */
    public static class Builder implements Interceptor.Builder {

        @Override
        public Interceptor build() {
            return new TypeInterceptor();
        }

        @Override
        public void configure(Context context) {

        }
    }

}
